package com.rpc.handler;

import com.google.protobuf.ByteString;
import com.netty.CIMSocketRequestHandler;
import com.netty.ImSocketServer;
import com.netty.constant.Constants;
import com.netty.model.proto.MessageBodyProto;
import com.netty.utils.DefaultTaskService;
import com.rpc.utils.ProtostuffUtils;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Random;
import java.util.stream.Collectors;

public class NettyServiceHandler implements CIMSocketRequestHandler {
    private Logger logger = LoggerFactory.getLogger(getClass());
    private ImSocketServer imSocketServer = null;
    private ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    public void startNetty(int port) throws InterruptedException {
        imSocketServer = new ImSocketServer(port,this,channelGroup);
        imSocketServer.startServer();
//        task();
    }

    public void task(){
        DefaultTaskService defaultTaskService = new DefaultTaskService();
        defaultTaskService.createTask("group", new Runnable() {
            @Override
            public void run() {
                List<Channel> serverChannel = channelGroup.stream().filter(item->item.attr(Constants.SessionConfig.UID).get().indexOf("server") > -1).collect(Collectors.toList());
                List<Channel> clientChannel = channelGroup.stream().filter(item->item.attr(Constants.SessionConfig.UID).get().startsWith("client")).collect(Collectors.toList());
                logger.info("链接数共计为："+channelGroup.size());
                serverChannel.stream().forEach(item->{
                    logger.info(item.attr(Constants.SessionConfig.UID).get());
                });
                logger.info("rpc客户端链接数为："+clientChannel.size()+"==="+"rpc服务端链接数为："+serverChannel.size());
            }
        });
    }
    @Override
    public void channelActive(Channel channel) {
        logger.info("有新连接接入"+channel.attr(Constants.SessionConfig.ID).get());
    }

    @Override
    public void channelRead0(ChannelHandlerContext channelHandlerContext, Object body) {
        if (body instanceof MessageBodyProto.Model) {
            String node = channelHandlerContext.channel().attr(Constants.SessionConfig.UID).get();
            if(channelHandlerContext.channel().attr(Constants.SessionConfig.UID).get().startsWith("client")){

                MessageBodyProto.Model model  = (MessageBodyProto.Model) body;
                String serverName = model.getDataMap().get("serverName").toStringUtf8();
                List<Channel> serverChannel = channelGroup.stream().filter(item->item.attr(Constants.SessionConfig.UID).get().startsWith(serverName+"server-")).collect(Collectors.toList());
                if(serverChannel.size()>0){
                    Channel channel = serverChannel.get(new Random().nextInt(serverChannel.size()));
                    channel.writeAndFlush(body);
                    System.out.println("请求路径为："+node+"===>"+channel.attr(Constants.SessionConfig.UID).get());
                }else{
                    MessageBodyProto.Model.Builder builder = MessageBodyProto.Model.newBuilder();
                    builder.setKey(model.getKey());
                    builder.setType(Constants.CmdType.MESSAGE);
                    builder.putAllData(model.getDataMap());
                    builder.putData("data", ByteString.copyFrom(ProtostuffUtils.serialize("没有可用服务")));
                    channelHandlerContext.channel().writeAndFlush(builder);
                }
            }else if((channelHandlerContext.channel().attr(Constants.SessionConfig.UID).get().indexOf("server") > -1)){
                MessageBodyProto.Model model = (MessageBodyProto.Model) body;
                String key = model.getKey();
                Channel channel = channelGroup.stream().filter(item -> item.attr(Constants.SessionConfig.UID).get().equals(key)).findAny().get();
                channel.writeAndFlush(body);
                System.out.println("请求路径为："+node+"===>"+channel.attr(Constants.SessionConfig.UID).get());
            }
        }
    }

    @Override
    public void channelUnregistered(Channel channel) {
        logger.info(channel.attr(Constants.SessionConfig.UID).get()+"断开");
    }
}
